#include "Stream.h"

#include <iostream>

#include "Meta.h"
#include "LogWriter.h"
#include "db/log_reader.h"
#include "leveldb/env.h"

using namespace leveldb;

Stream::InternalAppender::InternalAppender(Env * env, const std::string & name, uint64_t startPosition)
: env_(env)
, name_(name)
, writableFile_(nullptr)
, writer_(nullptr)
, startPosition_(startPosition)
{
    this->open();
}
Stream::InternalAppender::~InternalAppender(){
    this->close();
}

Status Stream::InternalAppender::append(const std::string & bytes){
    if(writer_ != nullptr)
        return writer_->AddRecord(bytes);
    return Status::InvalidArgument("writer_ is nullptr");
}

Status Stream::InternalAppender::open(){
    Status s = env_->NewAppendableFile(name_, &writableFile_); // create it if no-exist
    if(!s.ok()) 
        return s;

    writer_ = new log::NewWriter(writableFile_, startPosition_);
    return Status::OK();
}
void Stream::InternalAppender::close(){
    delete writer_; writer_ = nullptr;
    delete writableFile_;writableFile_ = nullptr;
}


Stream::InternalReader::InternalReader(Env * env, const std::string & name)
:env_(env)
, name_(name)
, readableFile_(nullptr)
, reader_(nullptr)
, isEof_(false)
, record_(new leveldb::Slice())
{
    this->open();
}

Stream::InternalReader::~InternalReader(){
    this->close();
    delete record_;
}

bool Stream::InternalReader::hasNext() const {
    return !isEof_;
}
void Stream::InternalReader::next() {
    if(false == isEof_){
        isEof_ = !(reader_->ReadRecord(record_, &scratch_));
    }
}

const Slice &  Stream::InternalReader::current() {
    return *record_;
}

//assert file is exist
Status Stream::InternalReader::open(){
    if(!env_->FileExists(name_)){
        isEof_ = true;
        return Status::OK();
    }
    //ASSERT( file exist == true);
    auto s = env_->NewSequentialFile(name_, &readableFile_);
    if(!s.ok()){
        return s;
    }

    struct LogReporter : public ::leveldb::log::Reader::Reporter {
    public:
        void Corruption(size_t bytes, const leveldb::Status & s) override {
            std::cout << "drops bytes:" << bytes << "status: " << s.ToString() << std::endl;
        }
    };
    
    LogReporter  reporter;
    reader_ = new log::Reader(readableFile_, &reporter, true, 0);

    if(false  == checkFileLength())
        isEof_ = true;
    else 
        isEof_ = !(reader_->ReadRecord(record_, &scratch_));
    return Status::OK();
}
void Stream::InternalReader::close(){
    delete reader_;  reader_ = nullptr;
    delete readableFile_; readableFile_ = nullptr;
}

bool Stream::InternalReader::checkFileLength(){
    assert(env_ != nullptr);
    uint64_t size = 0;
    if(env_->GetFileSize(name_, &size).ok()){
         return (size > log::kHeaderSize);
    }
    return false;
}


Stream::Stream(const std::string & name)
: env_(Env::Default())
, size_(new Size())
, name_(name) 
{
}

Stream::~Stream() {
    delete size_;
}

Size & Stream::size(){
    return *size_;
}
Env * Stream::env() {
    return env_;
}
const Size & Stream::size() const{
    return *size_;
}
const std::string Stream::name() const  {
    return name_;
}
bool Stream::exists() const{
    return env_->FileExists(name_);
}
leveldb::Status Stream::length(uint64_t & len) {
    return env_->GetFileSize(name_, &len);
}
leveldb::Status Stream::touch(){
    WritableFile * file = nullptr;
    auto s = env_->NewWritableFile(name_, &file);
    delete file;
    return s;
}
leveldb::Status Stream::destroy(){
    return env_->RemoveFile(name_);
}

Stream::InternalAppender * Stream::createAppender(uint64_t startPosition){
    return new Stream::InternalAppender(env_, name_, startPosition);
}
Stream::InternalReader * Stream::createReader(){
    return new Stream::InternalReader(env_, name_);
}
